package com.cloud.core.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkStreaming01_WordCount {


  val langs = Array("Scala", "Java", "Python", "Go", "JavaScript",
    "c", "r", "ruby", "rust", "php", "swift")

  val idx = 1

  def main(args: Array[String]): Unit = {

    // Should be some file on your system

    // TODO 创建环境对象
    val sparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("SparkStreaming_01")

    // 第二个参数表示批量处理的周期（采集周期）
    val ssc = new StreamingContext(sparkConf, Seconds(10))

    // TODO 逻辑处理
    // 获取监控文件目录
    val path = "hdfs://master:9000/wcinput/"

    val lang = langs(idx).toLowerCase()

    val lines = ssc.textFileStream(path)

    val words = lines.flatMap(_.split(" "))

    val wordToOne = words.map((_, 1))

    val wordToCount: DStream[(String, Int)] = wordToOne.reduceByKey(_ + _)

    val data = wordToCount.transform(rdd => {
      val dataRDD = rdd.sortBy(t => t._2, false)
      val sortResult = dataRDD.take(100)
      sortResult.foreach(println)
      dataRDD
    })

    data.print()

    data.saveAsTextFiles(lang, "log")

    // 由于SparkStreaming采集器是长期执行的任务，所以不能直接关闭
    // 如果main方法执行完毕，应用程序也会自动结束。所以不能让main执行完毕
    // 1. 启动采集器
    ssc.start()
    // 2. 等待采集器的关闭
    ssc.awaitTermination()

  }

}
